package com.learn.unify;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @create: 2023-04-17 16:31
 * @author: Mr.Du
 * --------------
 * @notes: 流批一体API
 **/
public class UnifyWordCount {
    public static void main(String[] args) throws Exception {
        //获取env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //env.setRuntimeMode(RuntimeExecutionMode.BATCH);//注意:使用DataStream实现批处理
        //env.setRuntimeMode(RuntimeExecutionMode.STREAMING);//注意:使用DataStream实现流处理
        //根据数据源自动选择使用流还是批
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        DataStream<String> lines = env.readTextFile("./data/input/wordcount.txt");

        lines.flatMap((String line, Collector<String> out) -> {
            String[] words = line.split(" ");
            for (String word : words) {
                out.collect(word);
            }
        }).returns(Types.STRING)
                .map(word-> Tuple2.of(word,1))
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(0)
                .sum(1)
                .print();

        env.execute("UnifyWordCount");
    }
}
